

package com.jacquesh;

import com.jacquesh.model.User;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * flatMap Demo
 * DataStream → DataStream
 * 遍历数据流中的每一个元素，产生N个元素 N=0，1，2
 */
public class FlinkDemo0001 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment=
                StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> stream= environment
                .socketTextStream("192.168.88.180",8888);
        DataStream<User> newStream= stream
                .flatMap(new FlatMapFunction<String, User>() {
            @Override
            public void flatMap(String s, Collector<User> collector)  {
                try {
                    String[] sp=s.split(",");
                    collector.collect(User.builder()
                            .id(Long.parseLong(sp[0]))
                            .name(sp[1])
                            .age(Integer.parseInt(sp[2]))
                            .build());
                }catch (Exception ex){
                    System.out.println(ex);
                }
            }
        }).setParallelism(2);

        newStream.print();
        environment.execute("1111");

    }
}
